//==============================================================================
// Copyright (c) 2008-2013 Niall Ryan. All Rights Reserved.
//==============================================================================

#ifndef HYDRA_HASHMAPS_LOCKFREEHASHMAP_H
#define HYDRA_HASHMAPS_LOCKFREEHASHMAP_H 1

#include "hydra/allocators/DelayedFreeAllocator.h"
#include "hydra/lists/LockFreeList.h"
#include "hydra/vectors/LockFreeVector.h"

namespace Hydra
{
	/**
	 * Lock-free hash map.
	 */
	template<typename K, typename T, typename Hash=HashUtil::HashFunction<K>, typename Pred=HashUtil::EqualTo<K>,
		typename Allocator=DelayedFreeAllocator>
	class LockFreeHashMap
	{
	public:
        static_assert(!Allocator::kAbaVulnerable, "Allocator must not be vulnerable to the ABA problem");

		LockFreeHashMap();
		~LockFreeHashMap();
		bool insert(const K& key, const T& value);
		bool erase(const K& key);
		bool contains(const K& key);
		bool find(const K& key, T* value);
		void setMaxLoadFactor(float maxLoadFactor) { m_maxLoadFactor = maxLoadFactor; }
		float getLoadFactor() const  { return static_cast<float>(m_size) / static_cast<float>(m_buckets.getSize()); }
		unsigned int getSize() const { return m_size; }
		bool isEmpty() const         { return (m_size == 0); }

	private:
		struct Pair
		{
			Pair() { }
			Pair(unsigned int reversedHash) : m_reversedHash(reversedHash) { }
			Pair(unsigned int reversedHash, const K& key, const T& value)
				: m_reversedHash(reversedHash), m_key(key), m_value(value) { }
			unsigned int m_reversedHash;
			K m_key;
			T m_value;
		};
		typedef LockFreeList<Pair, Allocator> PairList;
		typedef typename PairList::ConstIterator Bucket;
		typedef LockFreeVector<Atomic<Bucket*> > BucketVector;

		static unsigned int reverseBits(unsigned int v0);
		void initializeBucket(unsigned int bucketIndex);
		Bucket* getBucket(unsigned int bucketIndex);
		unsigned int makeSentinelReverseKey(unsigned int bucketIndex);
		unsigned int makeOrdinaryReverseKey(unsigned int hashValue);

		Allocator m_alloc;
		Atomic<unsigned int> m_size;
		Atomic<float> m_maxLoadFactor;

		PairList m_list;
		BucketVector m_buckets;
	};

	//==================================================================================================
	//==================================================================================================
	//==================================================================================================

	template<typename K, typename T, typename Hash, typename Pred, typename Allocator>
	inline LockFreeHashMap<K, T, Hash, Pred, Allocator>::LockFreeHashMap()
	{
		m_size = 0;
		m_maxLoadFactor = 2.0f;

		m_list.pushFront(Pair(makeSentinelReverseKey(0)));

		Atomic<Bucket*> newBucket;
		newBucket = static_cast<Bucket*>(m_alloc.alloc(sizeof(Bucket)));
		new(newBucket) Bucket(m_list);
		m_buckets.pushBack(newBucket);
	}

	template<typename K, typename T, typename Hash, typename Pred, typename Allocator>
	inline LockFreeHashMap<K, T, Hash, Pred, Allocator>::~LockFreeHashMap()
	{
	}

	template<typename K, typename T, typename Hash, typename Pred, typename Allocator>
	inline unsigned int LockFreeHashMap<K, T, Hash, Pred, Allocator>::reverseBits(unsigned int v0)
	{
		//23 operations, no branches, all simple operations
		//alternatives: ~16 operations, including 64-bit multiplies
		//              table lookups
		unsigned int v1 = ((v0 >> 1) & 0x55555555) | ((v0 & 0x55555555) << 1);
		unsigned int v2 = ((v1 >> 2) & 0x33333333) | ((v1 & 0x33333333) << 2);
		unsigned int v3 = ((v2 >> 4) & 0x0F0F0F0F) | ((v2 & 0x0F0F0F0F) << 4);
		unsigned int v4 = ((v3 >> 8) & 0x00FF00FF) | ((v3 & 0x00FF00FF) << 8);
		return (v4 >> 16) | (v4 << 16);
	}

	template<typename K, typename T, typename Hash, typename Pred, typename Allocator>
	inline unsigned int LockFreeHashMap<K, T, Hash, Pred, Allocator>::makeSentinelReverseKey(unsigned int bucketIndex)
	{
		//sentinels have lowest bit cleared, so they will sort before ordinary keys
		return reverseBits(bucketIndex) & 0xfffffffe;
	}

	template<typename K, typename T, typename Hash, typename Pred, typename Allocator>
	inline unsigned int LockFreeHashMap<K, T, Hash, Pred, Allocator>::makeOrdinaryReverseKey(unsigned int hashValue)
	{
		//ordinary keys have lowest bit set, so they will sort after sentinels
		return reverseBits(hashValue) | 0x01;
	}

	template<typename K, typename T, typename Hash, typename Pred, typename Allocator>
	inline void	LockFreeHashMap<K, T, Hash, Pred, Allocator>::initializeBucket(unsigned int bucketIndex)
	{
		//get an iterator to the parent bucket, this will provide the starting point for the search for the position
		// of the new sentinel
		//parent bucket is defined by clearing the highest set bit of the bucketIndex
		unsigned int parentBucketIndex = m_buckets.getSize();
		do 
		{
			parentBucketIndex >>= 1;
		} while( parentBucketIndex > bucketIndex);
		parentBucketIndex = bucketIndex - parentBucketIndex;

		Bucket* parentBucket = m_buckets[parentBucketIndex];
		if( !parentBucket )
		{
			initializeBucket(parentBucketIndex);
			parentBucket = m_buckets[parentBucketIndex];
		}
		assert(parentBucket);

		//insert the new sentinel into the list, or get the already existing sentinel
		unsigned int reversedSentinelKey = makeSentinelReverseKey(bucketIndex);
		PairList::Iterator iter; //this will point to the new sentinel when the while loop is complete
		while( true )
		{
			//start the search (again) from the parentBucket
			bool isFound = false;
			bool isFailedIteration = false;
			iter = parentBucket->getNextIterator();
			while( iter.hasNext() )
			{
				//check if we've found an already existing sentinel
				Pair pair = iter.peekNext();
				if( pair.m_reversedHash == reversedSentinelKey )
				{
					isFound = true;
					break;
				}
				//check if we've found the insertion point for the new sentinel, iter will be pointing to the element
				// after the desired sentinel position
				if( pair.m_reversedHash > reversedSentinelKey )
					break;
				//step forward, if the step fails then we must restart the search
				if( !iter.next(pair) )
				{
					isFailedIteration = true;
					break;
				}
			}
			if( isFound )
			{
				//iter is pointing to the already existing sentinel
				break;
			}
			else if( !isFailedIteration )
			{
				//attempt to insert the new sentinel, if successful the iter will be pointing to the new sentinel, if
				// insertion fails then we restart the search from the beginning
				if( m_list.insert(iter, Pair(reversedSentinelKey)) )
					break;
			}
		}

		//allocate the new bucket and attempt to set it in the bucket array
		Bucket* newBucket = static_cast<Bucket*>(m_alloc.alloc(sizeof(Bucket)));
		new(newBucket) Bucket(iter);
		Bucket* bucket = m_buckets[bucketIndex].compareAndSwap(newBucket, NULL);
		if( bucket )
		{
			//somebody beat us to it, that's ok, we'll just free our allocation and use theirs
			m_alloc.free(newBucket);
		}
	}

	template<typename K, typename T, typename Hash, typename Pred, typename Allocator>
	inline typename LockFreeHashMap<K, T, Hash, Pred, Allocator>::Bucket*
	LockFreeHashMap<K, T, Hash, Pred, Allocator>::getBucket(unsigned int bucketIndex)
	{
		Bucket* bucket = m_buckets[bucketIndex];
		if( !bucket )
		{
			initializeBucket(bucketIndex);
			bucket = m_buckets[bucketIndex];
		}
		return bucket;
	}

	template<typename K, typename T, typename Hash, typename Pred, typename Allocator>
	inline bool LockFreeHashMap<K, T, Hash, Pred, Allocator>::insert(const K& key, const T& value)
	{
		Hash hash;
		unsigned int hashValue = hash(key);
		unsigned int numBuckets = m_buckets.getSize();
		unsigned int bucketIndex = hashValue & (numBuckets-1);
		Bucket* bucket = getBucket(bucketIndex);

		//insert the new key into the list, or return if already present
		unsigned int reversedHash = makeOrdinaryReverseKey(hashValue);
		while( true )
		{
			//start the search (again) from the bucket
			PairList::Iterator iter = bucket->getNextIterator();
			bool isFailedIteration = false;
			while( iter.hasNext() )
			{
				//check if we've found an already existing key
				Pair pair = iter.peekNext();
				if( pair.m_reversedHash == reversedHash )
				{
					Pred pred;
					if( pred(pair.m_key, key) )
						return false;
				}
				//check if we've found the insertion point for the new key, iter will be pointing to the element
				// after the desired key position
				if( pair.m_reversedHash > reversedHash )
					break;
				//step forward, if the step fails then we must restart the search
				if( !iter.next(pair) )
				{
					isFailedIteration = true;
					break;
				}
			}
			if( !isFailedIteration )
			{
				//attempt to insert the new pair, if insertion fails then we must restart the search
				if( m_list.insert(iter, Pair(reversedHash, key, value)) )
				{
					++m_size;
					if( getLoadFactor() > m_maxLoadFactor )
						m_buckets.compareAndSwapSize(numBuckets, 2*numBuckets);
					return true;
				}
			}
		}
	}

	template<typename K, typename T, typename Hash, typename Pred, typename Allocator>
	inline bool LockFreeHashMap<K, T, Hash, Pred, Allocator>::erase(const K& key)
	{
		Hash hash;
		unsigned int hashValue = hash(key);
		unsigned int numBuckets = m_buckets.getSize();
		unsigned int bucketIndex = hashValue & (numBuckets-1);
		Bucket* bucket = getBucket(bucketIndex);

		//find the key in the list and remove it
		unsigned int reversedHash = makeOrdinaryReverseKey(hashValue);
		while( true )
		{
			//start the search (again) from the bucket
			PairList::Iterator iter = bucket->getNextIterator();
			bool isFailedIteration = false;
			while( iter.hasNext() )
			{
				//check if we've found an already existing key
				Pair pair = iter.peekNext();
				if( pair.m_reversedHash == reversedHash )
				{
					Pred pred;
					if( pred(pair.m_key, key) )
					{
						//attempt to remove the key from the list
						bool isReset;
						if( !m_list.remove(iter, isReset) )
						{
							//list removal failed, we must restart our search
							isFailedIteration = true;
							break;
						}
						--m_size;
						return true; //successful remove, we're finished
					}
				}
				//check if we've found the insertion point for the new key, iter will be pointing to the element
				// after the desired key position
				if( pair.m_reversedHash > reversedHash )
					break;
				//step forward, if the step fails then we must restart the search
				if( !iter.next(pair) )
				{
					isFailedIteration = true;
					break;
				}
			}
			if( !isFailedIteration )
				return false;
		}
	}

	template<typename K, typename T, typename Hash, typename Pred, typename Allocator>
	inline bool LockFreeHashMap<K, T, Hash, Pred, Allocator>::contains(const K& key)
	{
		Hash hash;
		unsigned int hashValue = hash(key);
		unsigned int numBuckets = m_buckets.getSize();
		unsigned int bucketIndex = hashValue & (numBuckets-1);
		Bucket* bucket = getBucket(bucketIndex);

		unsigned int reversedHash = makeOrdinaryReverseKey(hashValue);
		PairList::ConstIterator iter = *bucket;
		while( iter.hasNext() )
		{
			Pair pair = iter.next();
			if( pair.m_reversedHash == reversedHash )
			{
				Pred pred;
				if( pred(pair.m_key, key) )
					return true;
			}
			else if( pair.m_reversedHash > reversedHash )
				break;
		}
		return false;
	}

	template<typename K, typename T, typename Hash, typename Pred, typename Allocator>
	inline bool LockFreeHashMap<K, T, Hash, Pred, Allocator>::find(const K& key, T* value)
	{
		Hash hash;
		unsigned int hashValue = hash(key);
		unsigned int numBuckets = m_buckets.getSize();
		unsigned int bucketIndex = hashValue & (numBuckets-1);
		Bucket* bucket = getBucket(bucketIndex);

		unsigned int reversedHash = makeOrdinaryReverseKey(hashValue);
		PairList::ConstIterator iter = *bucket;
		while( iter.hasNext() )
		{
			Pair pair = iter.next();
			if( pair.m_reversedHash == reversedHash )
			{
				Pred pred;
				if( pred(pair.m_key, key) )
				{
					*value = pair.m_value;
					return true;
				}
			}
			else if( pair.m_reversedHash > reversedHash )
				break;
		}
		return false;
	}
}

#endif
